package com.atguigu.edu.realtime.dwd.db.app;

import com.atguigu.edu.realtime.common.base.BaseSQLApp;
import com.atguigu.edu.realtime.common.constant.Constant;
import com.atguigu.edu.realtime.common.util.SQLUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class DwdTradeOrderPaySucDetail extends BaseSQLApp {

    public static void main(String[] args) {
        new DwdTradeOrderPaySucDetail().start(
                10015,
                4,
                Constant.TOPIC_DWD_TRADE_ORDER_PAYMENT_SUCCESS
        );
    }
    @Override
    public void handle(StreamExecutionEnvironment env, StreamTableEnvironment tableEnv) {
        //从kafka的topic主题读取数据
        readOdsDB(tableEnv,Constant.TOPIC_DWD_TRADE_ORDER_PAYMENT_SUCCESS);
        //tableEnv.executeSql("select * from topic_db where `table` = 'payment_info'").print();
        //过滤出支付成功的数据
        Table paymentInfo = tableEnv.sqlQuery("select\n" +
                //"`type`,\n" +
                //"`old`['payment_status'],\n" +
                //"`data`['payment_status'],\n" +
                "data['id'] `id`,\n" +
                "data['out_trade_no'] `out_trade_no`,\n" +
                "data['order_id'] `order_id`,\n" +
                "data['alipay_trade_no'] `alipay_trade_no`,\n" +
                "data['total_amount'] `total_amount`,\n" +
                "data['trade_body'] `trade_body`,\n" +
                "data['payment_type'] `payment_type`,\n" +
                "data['payment_status'] `payment_status`,\n" +
                "data['create_time'] `create_time`,\n" +
                "data['update_time'] `update_time`,\n" +
                "data['callback_content'] `callback_content`,\n" +
                "data['callback_time'] `callback_time`,\n" +
                "`pt`,\n" +
                "ts,\n" +
                "et\n" +
                "from topic_db\n" +
                "where `table` = 'payment_info'\n" +
                "and `type` = 'insert'\n" +
                //"and `old`['payment_status'] is not null\n" +
                "and `data`['payment_status'] = '1602'"
        );
        tableEnv.createTemporaryView("payment_info",paymentInfo);
        //tableEnv.executeSql("select * from payment_info").print();
        //从下单表读取下单数据
        tableEnv.executeSql("create table dwd_trade_order_detail(\n" +
                "id string,\n" +
                "course_id string,\n" +
                "course_name string,\n" +
                "order_id string,\n" +
                "user_id string,\n" +
                "origin_amount string,\n" +
                "coupon_reduce string,\n" +
                "final_amount string,\n" +
                "create_time string,\n" +
                "date_id string,\n" +
                "order_status string,\n" +
                "out_trade_no string,\n" +
                "trade_body string,\n" +
                "session_id string,\n" +
                "province_id string,\n" +
                "ts bigint,\n" +
                "et as to_timestamp_ltz(ts, 0),\n" +
                "watermark for et as et - interval '3' second" +
                ")" + SQLUtil.getKafkaDDL(Constant.TOPIC_DWD_TRADE_ORDER_DETAIL,Constant.TOPIC_DWD_TRADE_ORDER_PAYMENT_SUCCESS));
        //tableEnv.executeSql("select * from dwd_trade_order_detail").print();

        //关联两个表数据
        Table resultTable = tableEnv.sqlQuery("select\n" +
                "pi.`id`,\n" +
                "pi.`alipay_trade_no`,\n" +
                "pi.`total_amount`,\n" +
                "pi.`payment_type`,\n" +
                "case when `payment_type` = '1101'\n" +
                "\t then '支付宝'\n" +
                "\t when `payment_type` = '1102'\n" +
                "\t then '微信'\n" +
                "\t when `payment_type` = '1103'\n" +
                "\t then '银联'\n" +
                "end `payment_type_name`,\t \n" +
                "pi.`payment_status`,\n" +
                "pi.`update_time`,\n" +
                "pi.`callback_content`,\n" +
                "pi.`callback_time`,\n" +
                "od.`id` `order_detail_id`,\n" +
                "od.`course_id`,\n" +
                "od.`course_name`,\n" +
                "od.`order_id`,\n" +
                "od.`user_id`,\n" +
                "od.`origin_amount`,\n" +
                "od.`coupon_reduce`,\n" +
                "od.`final_amount`,\n" +
                "od.`create_time`,\n" +
                "od.`date_id`,\n" +
                "od.`order_status`,\n" +
                "od.`out_trade_no`,\n" +
                "od.`trade_body`,\n" +
                "od.`session_id`,\n" +
                "od.`province_id`,\n" +
                "pi.ts\n" +
                "from payment_info pi\n" +
                "join dwd_trade_order_detail od\n" +
                "on pi.`order_id` = od.`order_id`\n" +
                "and od.`et` >= pi.`et` - interval '30' minute \n" +
                "and od.`et` <= pi.`et` + interval '5' second");
        //tableEnv.createTemporaryView("result_table",resultTable);
        //tableEnv.executeSql("select * from result_table").print();

        //将结果关联到kafka主题
        tableEnv.executeSql("create table dwd_trade_order_payment_success (\n" +
                "`id` string,\n" +
                "`order_id` string,\n" +
                "`alipay_trade_no` string,\n" +
                "`total_amount` string,\n" +
                "`payment_type` string,\n" +
                "`payment_type_name` string,\n" +
                "`payment_status` string,\n" +
                "`update_time` string,\n" +
                "`callback_content` string,\n" +
                "`callback_time` string,\n" +
                "`order_detail_id` string,\n" +
                "`course_id` string,\n" +
                "`course_name` string,\n" +
                "`user_id` string,\n" +
                "`origin_amount` string,\n" +
                "`coupon_reduce` string,\n" +
                "`final_amount` string,\n" +
                "`create_time` string,\n" +
                "`date_id` string,\n" +
                "`order_status` string,\n" +
                "`out_trade_no` string,\n" +
                "`trade_body` string,\n" +
                "`session_id` string,\n" +
                "`province_id` string,\n" +
                "ts bigint,\n" +
                "primary key(`order_detail_id`) not enforced\n" +
                ")" + SQLUtil.getUpsertKafkaDDL(Constant.TOPIC_DWD_TRADE_ORDER_PAYMENT_SUCCESS));

        //写入
        resultTable.executeInsert(Constant.TOPIC_DWD_TRADE_ORDER_PAYMENT_SUCCESS);

    }
}
